package com.example.transformation;

import com.example.model.WaterSensor;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * Created with IntelliJ IDEA.
 * ClassName: SocketSource
 * Package: com.example.source
 * Description:
 * User: fzykd
 *
 * @Author: LQH
 * Date: 2023-07-22
 * Time: 14:51
 */


//map filter flatMap 转换算子
public class SocketSource {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //设置线程数
        //设置平行数
        env.setParallelism(1);

        //单输出符运算符
        SingleOutputStreamOperator<WaterSensor> map = env.socketTextStream("hadoop102", 7777)
                //将读取的数据进行转换
                .map(new MapFunction<String, WaterSensor>() {
                    //map转换算子
                    @Override
                    public WaterSensor map(String value) throws Exception {
                        //传进来的字符串 切割为字符数组
                        String[] s = value.split(" ");
                        return new WaterSensor(s[0], Long.valueOf(s[1]), Integer.valueOf(2));
                    }
                });

        //提取WaterSensor的id
        //map.map(value -> value.getId()).print();
        //map 算子 一一映射 消费一个元素 生成一个元素 可以对元素进行加工


        //flatMap 扁平映射算子 主要是将数据流整体拆分为个体输出
        final SingleOutputStreamOperator<String> flatMap = map.flatMap(new FlatMapFunction<WaterSensor, String>() {
            //可以输出 不同的数据 消费一个元素 可以产生一个或者多个元素
            @Override
            public void flatMap(WaterSensor value, Collector<String> out) throws Exception {
                //输入的数据是 对象
                if ("s1".equals(value.getId())) {
                    //想下游发送数据
                    out.collect("id:" + value.getId() + "Ts" + value.getTs() + "flatMap算子");
                } else {
                    out.collect("id" + value.getId() + "Ts" + value.getTs() + "Vc" + value.getVc() + "flatMap算子");
                }

            }
        });

        flatMap.print();


        //过滤算子 从数据的数据中 提取id为s1的数据
        //因为返回的是boolean值 所以输入的数据和输出的数据 类型一致
        //过滤算子 也可以实现简单的分流操作 但是 一条数据会执行两次过滤操作
        SingleOutputStreamOperator<WaterSensor> filter = map.filter(new FilterFunction<WaterSensor>() {
            //过滤算子的返回值 固定是boolean值
            @Override
            public boolean filter(WaterSensor value) throws Exception {
                //过滤逻辑
                if ("s1".equals(value.getId())) {
                    //返回 true就是不过滤
                    return true;
                } else {
                    //false 就是过滤此数据 直接丢弃
                    return false;
                }
            }
        });

        filter.print();

        env.execute();
    }
}

/**
 * 总结 转换算子
 * map 消费一个元素 生成一个元素 对元素进行加工处理 输出类型 和输出类型可以不同
 * filter 过滤算子 输入类型和输出类型一致
 * flatMap 扁平映射 一个输入可以有多个输出 类型也可以不一样 向下游输出数据
 *
 */
